package function;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;


public class KafkaSinkFunction{
    public static FlinkKafkaProducer<JSONObject> getKafkaProducer(){
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.1.10:9092");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"kafka");
        FlinkKafkaProducer<JSONObject> kafkaProducer = new FlinkKafkaProducer<>("default_kafka", new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObject, @Nullable Long aLong) {
                System.out.println(jsonObject.getString("sinktable"));
                return new ProducerRecord<byte[], byte[]>(jsonObject.getString("sinktable"), jsonObject.getString("after").getBytes());
            }
        }, properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        return kafkaProducer;
    }
}
